package com.henan.provideservice.controller;

import com.henan.provideservice.api.response.TestResponse;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@RestController
@RequestMapping("/api/sse")
@CrossOrigin(origins = "*") // 允许所有跨域请求
@Slf4j
public class SseController {
    /**
     * 原始版本
     * @return
     */
//    @GetMapping(value = "/data", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//    public Flux<String> streamData() {
//        Flux<String> resonse = Flux.interval(Duration.ofSeconds(1)) // 每秒推送一次数据
//                .map(i -> "数据更新：" + LocalTime.now());
//        System.out.println("response="+ resonse);
//        return resonse;
//    }
//
//    @GetMapping(value = "/list", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//    public Flux<List<Integer>> streamList() {
//        Flux<List<Integer>> resonse = Flux.interval(Duration.ofSeconds(1))
//                .map(i -> IntStream.range(1, 6).mapToObj(n -> n * (int) (i + 1)).toList());
//        return resonse;
//    }

    /**
     * 解决报错问题
     * @return
     */
//    @GetMapping(value = "/data", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//    public Flux<String> streamData() {
//        return Flux.interval(Duration.ofSeconds(1))
//                .map(i -> "数据更新：" + LocalTime.now())
//                .onBackpressureDrop()  // 防止数据堆积导致异常
//                .doOnCancel(() -> log.info("SSE 连接已取消"))
//                .doOnError(error -> log.error("SSE 发生错误: {}", error.getMessage()))
//                .doFinally(signalType -> log.info("SSE 连接关闭: {}", signalType));
//    }
//
//    @GetMapping(value = "/list", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
//    public Flux<List<Integer>> streamList() {
//        return Flux.interval(Duration.ofSeconds(1))
//                .map(i -> IntStream.range(1, 6).mapToObj(n -> n * (int) (i + 1)).toList())
//                .onErrorResume(e -> {
//                    log.error("SSE list 发生错误: {}", e.getMessage());
//                    return Flux.empty();  // 让 Flux 继续运行
//                })
//                .onBackpressureDrop()
//                .doOnCancel(() -> log.info("SSE list 连接已取消"))
//                .doFinally(signalType -> log.info("SSE list 连接关闭: {}", signalType));
//    }

    /**
     * 模拟分页
     */
    @GetMapping(value = "/list", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<List<TestResponse>> streamPagedList() {
        return Flux.interval(Duration.ofSeconds(1)) // 每秒返回一次数据
                .map(i -> fetchData(i.intValue() * 500, 500)) // 每次获取 500 条数据
                .takeUntil(List::isEmpty) // 如果返回的数据为空，停止发送
                .onBackpressureDrop() // 避免数据堆积
                .doOnCancel(() -> log.info("SSE list 连接已取消"))
                .doFinally(signalType -> log.info("SSE list 连接关闭: {}", signalType));
    }

    // 模拟数据库分页查询
    private List<TestResponse> fetchData(int offset, int size) {
        List<TestResponse> allData = IntStream.range(1, 10000) // 假设总数据量 10000
                .mapToObj(n -> new TestResponse("Name_" + n, "Info_" + n))
                .toList();

        return allData.stream()
                .skip(offset) // 跳过 offset
                .limit(size)  // 获取 size 条数据
                .collect(Collectors.toList());
    }
}
